package com.lagou.flink;

import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.api.java.ExecutionEnvironment;
import org.apache.flink.api.java.operators.AggregateOperator;
import org.apache.flink.api.java.operators.DataSource;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.util.Collector;


/**
 * @author: yehw
 * @date: 2021/1/4 下午7:14
 * @description: java版Flink案例
 */
public class FirstFlink {

    public static void main(String[] args) throws Exception {
        //输入路径
        String inputPath="/Users/yehuaiwei/IdeaProjects/FirstFlink/input/wordCount.txt";
        String outPath="/Users/yehuaiwei/IdeaProjects/FirstFlink/outPut";
        //获得环境
        ExecutionEnvironment executionEnvironment = ExecutionEnvironment.getExecutionEnvironment();
        //读取文件
        DataSource<String> files = executionEnvironment.readTextFile(inputPath);
        //对数据进行处理
        AggregateOperator<Tuple2<String, Integer>> aggregateOperator = files.flatMap(new SplitSpace()).groupBy(0).sum(1);
        aggregateOperator.writeAsCsv(outPath,"\n"," ").setParallelism(1);
        executionEnvironment.execute();

    }

    static class SplitSpace implements FlatMapFunction<String, Tuple2<String,Integer>>{

        public void flatMap(String s, Collector<Tuple2<String, Integer>> collector) throws Exception {
            for (String word : s.split(" ")) {
                collector.collect(new Tuple2<String, Integer>(word,1));
            }
        }
    }

}
